package cn.itcast.czxy.BD18

import cn.itcast.czxy.BD18.bean.BaseMode
import org.apache.spark.sql.DataFrame

object Constellation extends BaseMode{
  override def setAppName: String = "Constellation"

  override def setLeven4Id: Int = 104

  override def getNewTag(leve5: DataFrame, hbaseDF: DataFrame): DataFrame = {
    import spark.implicits._
    val leven5 = leve5.map(row => {
      val id = row.getAs("id").toString
      val rule = row.getAs("rule").toString.split(";")
      (id, rule(0), rule(1))
    }).toDF("id", "start", "stop")

    val birth: DataFrame = hbaseDF.map(row => {
      val id = row.getAs("id").toString
      val birthdays = row.getAs("birthday").toString.split("-")
      (id, birthdays(1) + "-" + birthdays(2))
    }).toDF("id", "birthday")

    birth.join(leven5,birth("birthday").between(leven5("start"),leven5("stop"))).select(birth("id").as("userId"),leven5("id").as("tagsId"))
  }

  def main(args: Array[String]): Unit = {
    exec()
  }
}
